package com.china.produce.controller;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.lang.Pair;
import cn.hutool.core.thread.NamedThreadFactory;
import com.alibaba.fastjson.JSON;
import com.china.produce.pojo.User;
import com.google.common.collect.Lists;
import io.swagger.models.auth.In;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;


@RequestMapping("/completable")
@RestController
@Slf4j
public class CompletableFutureController {

    @RequestMapping("/testSupplyAsync")
    public String testSupplyAsync() {
        CompletableFuture<String> f1 = new CompletableFuture<>();
        CompletableFuture<String> f2 = new CompletableFuture<>();

        CompletableFuture<String> res1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
                f1.complete("this is first");
                log.info("11");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                return "11";
            }
        });

        CompletableFuture<String> res2 = CompletableFuture.supplyAsync(() -> {

            try {
                TimeUnit.SECONDS.sleep(5);
                f2.complete("this is second");
                log.info("22");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                return "22";
            }
        });

        res1.thenAcceptBoth(res2, (r1, r2) -> {

            try {
                TimeUnit.SECONDS.sleep(3);
                log.info("-------");
            }catch (Exception e) {

            }
            System.out.println(r1);

            System.out.println(r2);
        });


        return "hello";
    }

    @RequestMapping("/testJoin")
    public String testJoin() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "启动了 !");
            try {
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName() + "\t" + "结束了 !");
            } catch (Exception e) {
                e.printStackTrace();
            }
            return "testJoin -- success";
        });

        CompletableFuture<String>  future2 = future.whenComplete((r, e) -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "again !");
            if (e != null) {
                System.out.println("error !");
            }else {
                System.out.println("success :" + r);
            }
        });

        String result = future2.get();

        return result;
    }

    public static void main(String[] args) {


        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> 100).thenApply(s -> "future result :" + s);

        System.out.println(future.join());

        ArrayList<String> lists = Lists.newArrayList();

        lists.add("test1");
        lists.add("test2");

        CompletableFuture[] completableFutures = lists.stream().map(item -> supplyAsyncInner(item)).toArray(CompletableFuture[]::new);

        CompletableFuture.allOf(completableFutures).join();


        System.out.println("直接执行");
    }

    public static CompletableFuture<String> supplyAsyncInner(String param) {

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4, 10,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(1, true),
                new NamedThreadFactory("test-", true),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        return CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "启动" + " param " + param);
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (Exception e) {
                e.printStackTrace();
            }

            return "supplyAsync wait 10 seconds " + param;
        }, threadPoolExecutor).thenApply(res -> res + " test").handle(new BiFunction<String, Throwable, String>() {
            @Override
            public String apply(String s, Throwable throwable) {
                System.out.println(" a " + s);
                return  "res";
            }
        });
    }

    public static void main5(String[] args) throws ExecutionException, InterruptedException {

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4, 10,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(1, true),
                new NamedThreadFactory("test-", true),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        //supplyAsync 异步执行有返回值
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(5);
                System.out.println(Thread.currentThread().getName() + " step1 开始执行了");
            }catch (Exception e) {

            }finally {
                return "step1 start";
            }
        }, threadPoolExecutor);
        //supplyAsync 异步执行有返回值
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName() + " step2 开始执行了");
            }catch (Exception e) {

            }finally {
                return "step2 start";
            }
        }, threadPoolExecutor);
        //thenCombine 二个异步任务执行完成，组装参数其他supplyAsync的信息，成新的数据返回
        CompletableFuture<String> future3 = future1.thenCombine(future2, (res1, res2) -> {

            return "res1 :" + res1 + " res2 :" + res2;
        });

        System.out.println(future3.join());

        //allOf
        List<String> links = Arrays.asList("link1", "link2");
        List<CompletableFuture<String>> future4 = links.stream().map(link -> {

            return getLink(link);

        }).collect(Collectors.toList());

        CompletableFuture<Void> allFutures = CompletableFuture.allOf(future4.toArray(new CompletableFuture[links.size()]));
        //allOf没有返回值，所以thenApply的v为null
        CompletableFuture<List<String>> futures = allFutures.thenApply(v -> {
            System.out.println("v : " + v);
            return future4.stream().map(s -> s.join()).collect(Collectors.toList());
        });

        CompletableFuture<Long> futures2 = futures.thenApply(s -> {

            return s.stream().filter(p -> p.contains("link")).count();
        });

        System.out.println(futures2.join());
    }

    /**
     * 执行当前链接
     *
     * @param link
     * @return
     */
    public static CompletableFuture<String> getLink(String link) {

       return CompletableFuture.supplyAsync(() -> {
           try {
               TimeUnit.SECONDS.sleep(3);
               System.out.println("当前线程名称 ： " + Thread.currentThread().getName());
           }catch (Exception e) {

           }
           return "link " + link;
        });
    }

    public static void main31(String[] args) {
        CompletableFuture<User> future = CompletableFuture.supplyAsync(() -> {
            User user = new User();
            user.setPassword("123");
            user.setAccount("root");
            try {
                TimeUnit.SECONDS.sleep(15);
                log.info("1 current thread name : {}", Thread.currentThread().getName());
                return user;
            }catch (Exception e) {
                e.printStackTrace();
            }finally {
                return user;
            }
        });

        CompletableFuture<User> futures = future.whenComplete((r, e) -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            }catch (Exception es) {
                es.printStackTrace();
            }
            if (e != null) {
                e.printStackTrace();
            }else {
                log.info("2 current thread name : {} result : {}", Thread.currentThread().getName(), r);
            }
        });

        log.info("3 current thread name : {}", Thread.currentThread().getName());

        User user = futures.join();

        System.out.println(user);
    }

    public static void main3(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "启动了 !");
            try {
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName() + "\t" + "结束了 !");
            } catch (Exception e) {
                e.printStackTrace();
            }
            return "testJoin -- success";
        });

        CompletableFuture<String>  future2 = future.whenComplete((r, e) -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "again !");
            if (e != null) {
                System.out.println("error !");
            }else {
                System.out.println("success :" + r);
            }
        });

        CompletableFuture<String> future3 = future2.handle((r, e) -> {
            if (e != null) {
                System.out.println("error !");

                return r;
            } else {
                System.out.println("success : " + r);

                return r + " ..handle";
            }
        });

        String result = future3.join();

        System.out.println(result);
    }

    public static void main2(String[] args) {
        CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "启动");
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return 100;
        });

        System.out.println(f1.join());

        CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "启动");
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return 200;
        });

        System.out.println(f2.join());

        CompletableFuture<Integer> result = f1.thenCombine(f2, (x, y) -> {
            return x + y;
        });

        System.out.println(result.join());
    }

    public static void main1(String[] args) throws ExecutionException, InterruptedException {
        //step1: 异步执行
        CompletableFuture.runAsync(() -> {
            System.out.println("执行无返回任务开始---");
            try {
                TimeUnit.SECONDS.sleep(3);
            }catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("执行无返回任务结束---");
        });

        //step2: 异步执行，等待返回
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("执行返回任务开始时间" + DateUtil.now());
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello world";
        });
        System.out.println(future.get() +"\t"+"执行返回任务结束时间" + "\t" + DateUtil.now());

        //step3:异步执行、阻塞结果后对结果处理返回
        CompletableFuture<String> exceptionally = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("执行返回任务开始时间" + "\t" + DateUtil.now());
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello world";
        }).whenComplete((s, t) -> {
            //上面处理成功
            if (s != null) {
                s = s + " 。。。";
                System.out.println("处理: " + s);
            }
        }).exceptionally(new Function<Throwable, String>() {
            @Override
            public String apply(Throwable throwable) {
                return "异常结果";
            }
        });
        System.out.println("结果处理：" + exceptionally.get());


        ExecutorService executorService = Executors.newFixedThreadPool(5);

        for (int i = 0; i < 3; i++) {
            CompletableFuture<String> result = CompletableFuture.supplyAsync(() -> {
                String name = Thread.currentThread().getName();
                System.out.println(name + "执行开始时间" + DateUtil.now());
                long l = 5000l;
                try {
                    Thread.sleep(l);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return name;
            }, executorService).whenComplete((s, t) -> {
                System.out.println(s + "执行结束时间" + DateUtil.now());
            }).exceptionally(new Function<Throwable, String>() {
                @Override
                public String apply(Throwable throwable) {
                    System.out.println("exceptionally 异常" + throwable.getMessage());
                    return "666";
                }
            });

           System.out.println(result.get());
        }

        System.out.println("主线程"  + Thread.currentThread().getName() + "执行时间" + DateUtil.now());
    }
}
